package com.iot.processor;

import com.iot.processor.consumer.DataConsumer;
import com.iot.processor.producer.DataProducer;
import com.iot.processor.service.RedisService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * IoT数据处理系统主应用程序
 * 负责协调生产者、消费者和整个系统的运行
 */
public class Application {
    
    private static final Logger logger = LoggerFactory.getLogger(Application.class);
    
    // 系统配置
    private static final int QUEUE_CAPACITY = 10000;        // 消息队列容量
    private static final int PRODUCER_COUNT = 3;            // 生产者数量
    private static final int CONSUMER_COUNT = 5;            // 消费者数量
    private static final int MESSAGES_PER_SECOND = 100;     // 每个生产者每秒生成消息数
    private static final int DEVICE_COUNT_PER_PRODUCER = 100; // 每个生产者负责的设备数量
    
    // 系统组件
    private BlockingQueue<String> messageQueue;
    private RedisService redisService;
    private ExecutorService executorService;
    private ScheduledExecutorService scheduledExecutor;
    
    private List<DataProducer> producers;
    private List<DataConsumer> consumers;
    
    private volatile boolean systemRunning = false;
    
    public static void main(String[] args) {
        Application app = new Application();
        app.start();
    }
    
    /**
     * 启动应用程序
     */
    public void start() {
        logger.info("=== IoT数据处理系统启动 ===");
        
        try {
            // 初始化系统组件
            initializeComponents();
            
            // 测试Redis连接
            if (!testRedisConnection()) {
                logger.error("Redis连接失败，系统无法启动");
                return;
            }
            
            // 启动生产者和消费者
            startProducersAndConsumers();
            
            // 启动监控任务
            startMonitoring();
            
            // 启动用户交互界面
            startUserInterface();
            
        } catch (Exception e) {
            logger.error("系统启动失败", e);
        } finally {
            shutdown();
        }
    }
    
    /**
     * 初始化系统组件
     */
    private void initializeComponents() {
        logger.info("初始化系统组件...");
        
        // 创建消息队列
        messageQueue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
        
        // 创建Redis服务
        redisService = new RedisService();
        
        // 创建线程池
        executorService = Executors.newFixedThreadPool(PRODUCER_COUNT + CONSUMER_COUNT);
        scheduledExecutor = Executors.newScheduledThreadPool(2);
        
        // 创建生产者列表
        producers = new ArrayList<>();
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            int deviceIdStart = i * DEVICE_COUNT_PER_PRODUCER + 1;
            int deviceIdEnd = (i + 1) * DEVICE_COUNT_PER_PRODUCER;
            
            DataProducer producer = new DataProducer(
                messageQueue, 
                "Producer-" + (i + 1), 
                deviceIdStart, 
                deviceIdEnd, 
                MESSAGES_PER_SECOND
            );
            producers.add(producer);
        }
        
        // 创建消费者列表
        consumers = new ArrayList<>();
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            DataConsumer consumer = new DataConsumer(
                messageQueue, 
                redisService, 
                "Consumer-" + (i + 1)
            );
            consumers.add(consumer);
        }
        
        logger.info("系统组件初始化完成 - 生产者: {}, 消费者: {}, 队列容量: {}", 
                   PRODUCER_COUNT, CONSUMER_COUNT, QUEUE_CAPACITY);
    }
    
    /**
     * 测试Redis连接
     */
    private boolean testRedisConnection() {
        logger.info("测试Redis连接...");
        boolean connected = redisService.testConnection();
        if (connected) {
            logger.info("Redis连接测试成功");
            logger.info(redisService.getPoolStatus());
        } else {
            logger.error("Redis连接测试失败，请检查Redis服务是否启动");
        }
        return connected;
    }
    
    /**
     * 启动生产者和消费者
     */
    private void startProducersAndConsumers() {
        logger.info("启动生产者和消费者...");
        
        systemRunning = true;
        
        // 启动所有生产者
        for (DataProducer producer : producers) {
            executorService.submit(producer);
        }
        
        // 启动所有消费者
        for (DataConsumer consumer : consumers) {
            executorService.submit(consumer);
        }
        
        logger.info("所有生产者和消费者已启动");
    }
    
    /**
     * 启动系统监控
     */
    private void startMonitoring() {
        logger.info("启动系统监控...");
        
        // 每30秒输出一次系统状态
        scheduledExecutor.scheduleAtFixedRate(this::printSystemStatus, 30, 30, TimeUnit.SECONDS);
        
        // 每5分钟输出一次详细统计
        scheduledExecutor.scheduleAtFixedRate(this::printDetailedStats, 300, 300, TimeUnit.SECONDS);
    }
    
    /**
     * 启动用户交互界面
     */
    private void startUserInterface() {
        logger.info("启动用户交互界面，输入命令进行控制:");
        logger.info("命令列表:");
        logger.info("  status  - 显示系统状态");
        logger.info("  stats   - 显示详细统计");
        logger.info("  redis   - 显示Redis状态");
        logger.info("  stop    - 停止系统");
        logger.info("  help    - 显示帮助信息");
        
        Scanner scanner = new Scanner(System.in);
        
        while (systemRunning) {
            try {
                System.out.print("IoT-Processor> ");
                String command = scanner.nextLine().trim().toLowerCase();
                
                switch (command) {
                    case "status":
                        printSystemStatus();
                        break;
                    case "stats":
                        printDetailedStats();
                        break;
                    case "redis":
                        printRedisStatus();
                        break;
                    case "stop":
                        logger.info("收到停止命令，正在关闭系统...");
                        systemRunning = false;
                        break;
                    case "help":
                        printHelpInfo();
                        break;
                    case "":
                        // 空命令，忽略
                        break;
                    default:
                        System.out.println("未知命令: " + command + "，输入 'help' 查看帮助");
                        break;
                }
                
            } catch (Exception e) {
                logger.error("处理用户命令时发生异常", e);
            }
        }
        
        scanner.close();
    }
    
    /**
     * 打印系统状态
     */
    private void printSystemStatus() {
        logger.info("=== 系统状态 ===");
        logger.info("队列大小: {}/{}", messageQueue.size(), QUEUE_CAPACITY);
        
        // 生产者状态
        long totalProduced = producers.stream().mapToLong(DataProducer::getMessageCount).sum();
        logger.info("生产者状态 - 总生产消息: {}", totalProduced);
        
        // 消费者状态
        long totalProcessed = consumers.stream().mapToLong(c -> c.getStats().getProcessedCount()).sum();
        long totalSuccess = consumers.stream().mapToLong(c -> c.getStats().getSuccessCount()).sum();
        long totalError = consumers.stream().mapToLong(c -> c.getStats().getErrorCount()).sum();
        
        double overallSuccessRate = totalProcessed > 0 ? (double) totalSuccess / totalProcessed * 100 : 0;
        
        logger.info("消费者状态 - 总处理: {}, 成功: {}, 失败: {}, 成功率: {:.2f}%", 
                   totalProcessed, totalSuccess, totalError, overallSuccessRate);
    }
    
    /**
     * 打印详细统计信息
     */
    private void printDetailedStats() {
        logger.info("=== 详细统计信息 ===");
        
        // 生产者详细信息
        logger.info("生产者详情:");
        for (DataProducer producer : producers) {
            logger.info("  {}", producer.getStatus());
        }
        
        // 消费者详细信息
        logger.info("消费者详情:");
        for (DataConsumer consumer : consumers) {
            logger.info("  {}", consumer.getStatus());
        }
    }
    
    /**
     * 打印Redis状态
     */
    private void printRedisStatus() {
        logger.info("=== Redis状态 ===");
        logger.info("连接状态: {}", redisService.testConnection() ? "正常" : "异常");
        logger.info(redisService.getPoolStatus());
    }
    
    /**
     * 打印帮助信息
     */
    private void printHelpInfo() {
        System.out.println("=== 命令帮助 ===");
        System.out.println("status  - 显示系统运行状态和基本统计信息");
        System.out.println("stats   - 显示所有生产者和消费者的详细统计信息");
        System.out.println("redis   - 显示Redis连接状态和连接池信息");
        System.out.println("stop    - 优雅停止整个系统");
        System.out.println("help    - 显示此帮助信息");
    }
    
    /**
     * 关闭系统
     */
    private void shutdown() {
        logger.info("正在关闭系统...");
        
        // 停止所有生产者
        for (DataProducer producer : producers) {
            producer.stop();
        }
        
        // 停止所有消费者
        for (DataConsumer consumer : consumers) {
            consumer.stop();
        }
        
        // 关闭线程池
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        
        if (scheduledExecutor != null) {
            scheduledExecutor.shutdown();
        }
        
        // 关闭Redis连接
        if (redisService != null) {
            redisService.close();
        }
        
        logger.info("系统已关闭");
    }
}